package elastic7

import (
	"context"
	"go2/log"
	"net/http"
	"runtime"
	"sync"
	time2 "time"

	"github.com/olivere/elastic/v7"
)

type ElasticRetrier struct {
	backoff elastic.Backoff
}

func NewElasticRetrier(initialTimeout, maxTimeout time2.Duration) *ElasticRetrier {
	return &ElasticRetrier{
		backoff: elastic.NewExponentialBackoff(initialTimeout, maxTimeout),
	}
}

func (c *ElasticRetrier) Retry(ctx context.Context, retry int, req *http.Request, resp *http.Response, err error) (time2.Duration, bool, error) {
	log.Errorf("es failed[%d]: %v", retry, err)

	// 让退让策略决定等待多长时间以及是否停止
	wait, stop := c.backoff.Next(retry)
	return wait, stop, nil
}

type Elastic struct {
	m_database_id uint64
	m_cli         *elastic.Client
	m_bp          *elastic.BulkProcessor
}

func (c *Elastic) Flush() {
	if c.m_bp == nil {
		return
	}
	c.m_bp.Flush()
}

func (c *Elastic) GetSates() elastic.BulkProcessorStats {
	if c.m_bp == nil {
		return elastic.BulkProcessorStats{}
	}
	return c.m_bp.Stats()
}

func (c *Elastic) Insert(table_name, id string, data any) {
	if c.m_bp == nil {
		return
	}
	c.m_bp.Add(elastic.NewBulkIndexRequest().Index(table_name).Id(id).Doc(data))
}

func (c *Elastic) Update(table_name, id string, data any) {
	if c.m_bp == nil {
		return
	}
	c.m_bp.Add(elastic.NewBulkUpdateRequest().Index(table_name).Id(id).Doc(data))
}

func (c *Elastic) beforeCallback(executionId int64, requests []elastic.BulkableRequest) {
}

func (c *Elastic) afterCallback(executionId int64, requests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
	if err != nil {
		log.Errorf("es error0: %v", err)
		runtime.Gosched()
	}

	if response != nil && response.Errors {
		for _, value := range response.ByAction("update") {
			if value.Status != 404 && value.Status != 200 {
				log.Errorf("es response error1: %v", value)
				runtime.Gosched()
			}
		}
		for _, value := range response.ByAction("create") {
			log.Errorf("es response error2: %v", value)
			runtime.Gosched()
		}
	}

	for _, req := range requests {
		if _, err := req.Source(); err != nil {
			log.Debugf("es error1: %v", err)
			runtime.Gosched()
		}
	}
}

var (
	m_elastic_sync sync.Mutex
	mp_elastic_map map[uint64]*Elastic
	m_client       *elastic.Client
)

func (c *Elastic) GetID(index, id string) (map[string]any, error) {
	// 根据id查询文档
	get1, err := m_client.Get().
		Index(index).            // 指定索引名
		Id(id).                  // 设置文档id
		Do(context.Background()) // 执行请求
	if get1.Found {
		return get1.Fields, nil
	}
	return nil, err
}

func InitElastic(database_id uint64, works int, counts int, time time2.Duration, options ...elastic.ClientOptionFunc) {
	m_elastic_sync.Lock()
	defer m_elastic_sync.Unlock()

	if mp_elastic_map == nil {
		mp_elastic_map = make(map[uint64]*Elastic)
	}

	if cli, ok := mp_elastic_map[database_id]; ok && cli != nil {
		return
	}

	var err error

	for {
		m_client, err = elastic.NewSimpleClient(options...)
		if err == nil {
			// Setup a bulk processor
			for {
				elastic_client := &Elastic{m_cli: m_client, m_database_id: database_id}
				p, err := m_client.BulkProcessor().
					Name("MyBackgroundWorker-1").
					Before(elastic_client.beforeCallback). // func to call before commits
					After(elastic_client.afterCallback).   // func to call after commits
					Workers(works).
					BulkActions(counts). // commit if # requests >= 1000
					//BulkSize(1 << 20).              // commit if size of requests >= 2 MB
					FlushInterval(time). // commit every 30s
					Stats(true).         // collect stats
					Do(context.Background())
				if err == nil && p != nil {
					elastic_client.m_bp = p
					mp_elastic_map[database_id] = elastic_client
					return
				}
				log.Errorf("es build processor init failed: %v", err)
				time2.Sleep(100 * time2.Millisecond)
			}
		}
		log.Errorf("es init failed: %v", err)
		time2.Sleep(1 * time2.Second)
	}
}

// GetESClient 获取 ES client
func GetElastic(id uint64) *Elastic {
	if mp_elastic_map == nil {
		return nil
	}
	return mp_elastic_map[id]
}

func FlushElastic() {
	for _, cli := range mp_elastic_map {
		cli.Flush()
	}
}
